package com.hzh.SparkStreaming

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}

import java.sql.{Connection, DriverManager, PreparedStatement}
import java.text.SimpleDateFormat
import java.util.Date

object Demo8Card {
  def main(args: Array[String]): Unit = {

    /**
     * 创建环境
     *
     */

    val spark: SparkSession = SparkSession
      .builder()
      .config("spark.sql.shuffle.partitions", 1)
      .master("local[2]")
      .appName("Demo4DSToRDDAndDF")
      .getOrCreate()

    val sc: SparkContext = spark.sparkContext

    val ssc = new StreamingContext(sc, Durations.seconds(5))

    val sscDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)

    /**
     * 1、解析json格式的数据
     */

    val carDS: DStream[(Long, (Double, Int))] = sscDS.map(line => {
      //使用fastjson工具解析json格式的数据
      val cardJson: JSONObject = JSON.parseObject(line)
      //取出卡口编号和车速
      val card: Long = cardJson.getLong("card")
      val speed: Double = cardJson.getDouble("speed")

      //返回元组
      (card, (speed, 1))
    })

    /**
     * 2、实时统计每个卡口的平均车速，和车的平均速度
     *
     */
    val sum_speed_and_num: DStream[(Long, (Double, Int))] = carDS.reduceByKey((kv1: (Double, Int), kv2: (Double, Int)) => {
      //计算 速度
      val speed: Double = kv1._1 + kv2._1
      //计算车数量
      val num: Int = kv1._2 + kv2._2
      (speed, num)
    })


    /**
     * 3、计算平均速度
     *
     */
    val avg_num_ds: DStream[(Long, Int, Double)] = sum_speed_and_num.map {
      case (card: Long, (sumSpeed: Double, num: Int)) =>
        val avg_speed: Double = sumSpeed / num
        (card, num, avg_speed)
    }

    /**
     *
     * 4、将统计的结果保存到MySQL中
     *
     * 将数据保存到数据库中存在的问题
     *
     * 1、如果直接使用foreach，会为每一个数据创建一个连接，效率低，而且会导致数据库压力过大
     * 2、如果将网络连接放在foreach算子外面，会报错，连接不能被序列化（网络连接不能在网络中传输）
     *
     * 正确写法：使用foreachPartition，指挥为每一个分区创建连接
     * RDD的foreach和foreachPartition的区别
     *
     * RDD的foreach一次处理一条数据
     * foreachPartition一次处理一个分区的数据
     *
     */

    avg_num_ds.foreachRDD(rdd => {

      //获取统计的时间
      val date = new Date()

      //格式化数据
      val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

      val strDate: String = sdf.format(date)

      //循环数据，将数据一条一条的保存到数据库中，使用JDBC
      rdd.foreachPartition(iter => {

        //加载驱动
        Class.forName("com.mysql.jdbc.Driver")

        //2、创建连接
        val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/test", "root", "123456")

        //3、编写插入数据的SQL
        val statement: PreparedStatement = conn.prepareStatement("insert into avg_speed_num(card,com_date,num,avg_speed) values(?,?,?,?)")

        //迭代器的一个方法，不是算子
        iter.foreach {
          case (card: Long, num: Int, avg_speed: Double) =>
            //设置参数
            statement.setLong(1, card)
            statement.setString(2, strDate)
            statement.setInt(3, num)
            statement.setDouble(4, avg_speed)
            //插入数据
            statement.execute()
        }
        statement.close()
        conn.close()
      })
    })

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()
  }
}
